package com.inspur2

import org.apache.spark.{SparkConf, SparkContext}

import scala.io.Source

object DataClear {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      //      .setMaster("spark://192.168.66.88:7077")
      .setMaster("local")
      .setAppName("data clear")
    val sc = new SparkContext(conf)

    var mapping:Map[String, String] = Map()
    var eastList:List[String] = List()
    var westList:List[String] = List()

    val source_mapping = Source.fromFile("C:\\Software\\idea\\Work\\SparkB\\src\\main\\java\\com\\inspur2\\NBA\\mappings.txt")
    val source_east = Source.fromFile("C:\\Software\\idea\\Work\\SparkB\\src\\main\\java\\com\\inspur2\\NBA\\east.txt")
    val source_west = Source.fromFile("C:\\Software\\idea\\Work\\SparkB\\src\\main\\java\\com\\inspur2\\NBA\\west.txt")

    for(line<-source_mapping.getLines()) {
      val arr = line.split(",")
      mapping += (arr(0) -> arr(1)) //旧队名 -> 新队名
    }
    source_mapping.close()

    for(line<-source_east.getLines()) {
      eastList = line :: eastList
    }
    source_east.close()

    for(line<-source_west.getLines()) {
      westList = line :: westList
    }
    source_west.close()

    val bc_mapping = sc.broadcast(mapping)
    val bc_east = sc.broadcast(eastList)
    val bc_west = sc.broadcast(westList)

    val nba = sc.textFile("hdfs://192.168.66.88:8020/0622/data/nba")

    nba
      .map(x => x.split(","))
      .map(x => {
        if(bc_mapping.value.contains(x(2)))
          x(2) = mapping(x(2))

        if(bc_mapping.value.contains(x(4)))
          x(4) = mapping(x(4))

        var result:Array[String] = Array()
        result = result :+ x(0) :+ x(1) :+ x(2) :+ x(3) :+ x(4)

        if(x(0).toInt >= 1970) { //1970年后东西分区
          var flag = ""
          if(bc_east.value.contains(x(2)))
            flag = "East"
          if(bc_west.value.contains(x(2)))
            flag = "West"
          result = result :+ flag
        }

        result.mkString(",")
      })
      .repartition(1)
      .saveAsTextFile("hdfs://192.168.66.88:8020/0622/clear_data")

    sc.stop()

  }

}